package com.test45_JUC;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

/**
 * Phaser，同步处理工具类，主要用于分阶段的任务处理上
 * JDK1.7引入
 */
public class Demo32_Phaser {

    public static void main(String[] args) throws Exception {
        // test1();
        // test2();
        // test3();
        test4();
    }

    /**
     * Phaser代替CyclicBarrier
     */
    public static void test1() throws Exception {
        Phaser phaser = new Phaser(2);  // 定义2个任务
        System.out.println("【Phaser阶段1】" + phaser.getPhase());  // 初始阶段
        for (int x = 0; x < 2; x++) {
            TimeUnit.SECONDS.sleep(1);
            new Thread(() -> {
                System.out.printf("【%s】我已就位，等待下一步的执行命令。%n", Thread.currentThread().getName());
                phaser.arriveAndAwaitAdvance();  // Phaser就位，拥有线程等待
                System.out.printf("【%s】人员齐备，准备执行新的任务。%n", Thread.currentThread().getName());
            }, "士兵 - " + x).start();
        }
        TimeUnit.SECONDS.sleep(3);
        System.out.println("【Phaser阶段2】" + phaser.getPhase());  // 初始阶段
    }

    /**
     * Phaser代替CountDownLatch
     */
    public static void test2() throws Exception {
        Phaser phaser = new Phaser(2);  // 定义2个任务
        for (int x = 0; x < 2; x++) {
            TimeUnit.SECONDS.sleep(1);
            new Thread(() -> {
                System.out.printf("【%s】达到已经上车。%n", Thread.currentThread().getName());
                phaser.arrive();    // 等价于countDown()方法
            }, "游客 - " + x).start();
        }
        phaser.awaitAdvance(phaser.getPhase());  // 等待达到阶段
        System.out.println("【主线程】人齐了，开车走人，下一个景点购物消费。");
    }

    /**
     * Phaser控制任务执行轮数
     */
    public static void test3() throws Exception {
        int repeat = 2;  // 执行的轮数配置
        Phaser phaser = new Phaser() {  // 进行一些处理方法的覆写
            // 回调处理
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.printf("【onAdvance()处理】进阶处理操作，phase = %d、registeredParties = %d%n", phase, registeredParties);
                return phase + 1 >= repeat || registeredParties == 0;   // 终止处理
            }
        };
        for (int x = 0; x < 2; x++) {
            phaser.register();  // 注册参与者的线程
            new Thread(() -> {
                while (!phaser.isTerminated()) {     // 现在没有终止Phaser执行
                    phaser.arriveAndAwaitAdvance();  // 等待其他的线程就位
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.printf("【%s】YOOTK业务处理。%n", Thread.currentThread().getName());
                }
            }, "子线程 - " + x).start();
        }
    }

    /**
     * Phaser任务层级划分
     */
    public static final int THRESHOLD = 2;  // 阈值，每个Phaser对应的任务数

    public static void test4() throws Exception {
        int repeat = 2;  // 定义任务的重复周期
        Phaser phaser = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.printf("【onAdvance()处理】进阶处理操作，phase = %d、registeredParties = %d%n", phase, registeredParties);
                return phase + 1 >= repeat || registeredParties == 0;   // 终止处理
            }
        };
        Tasker[] taskers = new Tasker[5];   // 创建5个子任务
        build(taskers, 0, taskers.length, phaser);  // 构件Phaser层级
        for (int x = 0; x < taskers.length; x++) {
            new Thread(taskers[x], "子线程 - " + x).start();
        }
    }

    // 构件Phaser层级
    private static void build(Tasker[] taskers, int low, int high, Phaser parent) {
        if ((high - low) > THRESHOLD) {
            for (int x = low; x < high; x += THRESHOLD) {
                int limit = Math.min(x + THRESHOLD, high);
                build(taskers, x, limit, new Phaser(parent));
            }
        } else {
            for (int x = low; x < high; x++) {
                taskers[x] = new Tasker(parent);
            }
        }
    }
}

/**
 * 定义任务的处理线程
 */
class Tasker implements Runnable {
    private Phaser phaser;

    public Tasker(Phaser phaser) {
        this.phaser = phaser;
        this.phaser.register();  // 将当前线程进行注册
    }

    @Override
    public void run() {
        while (!phaser.isTerminated()) {    // 任务还在执行
            this.phaser.arriveAndAwaitAdvance();    // 等待其他参与者的线程
            System.out.printf("【%s】YOOTK业务处理。%n", Thread.currentThread().getName());
        }
    }
}
